import json
import logging
import time
from functools import wraps
from typing import Callable, TypeVar, cast

from elasticsearch6 import helpers, Elasticsearch
from flask import Blueprint, current_app, request
from flask import jsonify

from airflow.patsnap.service import project_service
from airflow.configuration import conf
from airflow.exceptions import AirflowException
from airflow.models import DagPickle, TaskInstance
from airflow.models.idata_project import Project
from airflow.www.api import provide_user

log = logging.getLogger(__name__)

T = TypeVar("T", bound=Callable)  # pylint: disable=invalid-name


def requires_authentication(function: T):
    """Decorator for functions that require authentication"""

    @wraps(function)
    def decorated(*args, **kwargs):
        return current_app.api_auth.requires_authentication(function)(*args, **kwargs)

    return cast(T, decorated)


api_speech_project = Blueprint('api_speech_project', __name__)


# user route
@api_speech_project.route('/projects/<int:id>', methods=['GET'])
def get_project(id):
    try:
        project = Project.get_project(id)
        return jsonify(code=0, message='ok', data=Project.to_json(project))
    except AirflowException as e:
        log.error("get_project error =%s, params=%s", e, id, )
        return jsonify(code=500, message=str(e), data=None)


# 获取所有项目
@api_speech_project.route('/projects', methods=['GET'])
def get_projects():
    try:
        page = request.args.get('page', 0, type=int)
        result = project_service.get_projects(page)
        return jsonify(code=0, message='ok', data=result)
    except AirflowException as e:
        log.error("get_projects error =%s", e, )
        return jsonify(code=500, message=str(e), data=None)


@api_speech_project.route('/projects', methods=['POST'])
@provide_user
def add_project(oa_user_name):
    try:
        project_name = request.args.get("project_name", type=str, default=None)
        project_desc = request.args.get("project_desc", type=str, default=None)
        project_status = request.args.get("project_status", type=int, default=None)
        pool = request.args.get("pool", type=str, default='default')

        if project_name is None or project_desc is None or project_status is None:
            log.error(
                "add_project params error project_name={%s},project_desc={%s},project_status={%s},pool={%s}",
                project_name, project_desc, project_status,pool)
            return jsonify(code=400, message="add_projects params has None,", data=None)
        project_service.add_project(project_name, project_desc, project_status, oa_user_name, pool)
        return jsonify(code=0, message='ok', data=None)
    except AirflowException as e:
        log.error("add_project error =%s", e)
        return jsonify(code=500, message=str(e), data=None)


@api_speech_project.route('/projects/<int:id>', methods=['DELETE'])
def delete_project(id):
    try:
        project_service.delete_project(id)
        return jsonify(code=0, message='ok', data=None)
    except AirflowException as e:
        log.error("delete_project error =%s", e)
        return jsonify(code=500, message=str(e), data=None)


@api_speech_project.route('/projects', methods=['PUT'])
@provide_user
def update_project(oa_user_name):
    try:
        project_id = request.args.get("id", type=int, default=None)
        project_name = request.args.get("project_name", type=str, default=None)
        project_desc = request.args.get("project_desc", type=str, default=None)
        project_status = request.args.get("project_status", type=str, default=None)
        pool = request.args.get("pool", type=str, default=None)

        if project_id is None or project_name is None or project_desc is None or project_status is None:
            log.error(
                "update_project params error id={%s},user_name={%s},oa_user_name={%s},department_name={%s},pool={%s}",
                project_id, project_name, project_desc, project_status, pool)
            return jsonify(code=400, message="update_project params has None,", data=None)
        project_service.update_project(project_id, project_name, project_desc, project_status, oa_user_name, pool)
        return jsonify(code=0, message='ok', data=None)
    except AirflowException as e:
        log.error("update_project error =%s", e)
        return jsonify(code=500, message=str(e), data=None)


@api_speech_project.route('/projects/<int:id>/users', methods=['GET'])
def get_project_users(id):
    try:
        page = request.args.get('page', 0, type=int)
        result = project_service.get_project_users(id, page)
        return jsonify(code=0, message='ok', data=result)
    except AirflowException as e:
        log.error("get_project_users error =%s, params=%s", e, id, )
        return jsonify(code=500, message=str(e), data=None)


@api_speech_project.route('/projects/users/', methods=['POST'])
@provide_user
def add_project_users(oa_user_name):
    try:
        project_id = request.args.get("project_id", type=int, default=None)
        users = request.json

        if project_id is None:
            log.error("add_project_user params error project_id={%s}", project_id)
            return jsonify(code=400, message="get_projects_user params has None,", data=None)

        result = project_service.add_project_users(oa_user_name, project_id, users)
        return jsonify(code=0, message='ok', data=result)
    except AirflowException as e:
        log.error("add_project_users error =%s", e, )
        return jsonify(code=500, message=str(e), data=None)


@api_speech_project.route('/projects/<int:id>/users', methods=['DELETE'])
def delete_project_users(id):
    try:
        users = request.json
        result = project_service.delete_project_users(id, users)
        return jsonify(code=0, message='ok', data=result)
    except AirflowException as e:
        log.error("delete_project_users error =%s", e, )
        return jsonify(code=500, message=str(e), data=None)


@api_speech_project.route('/projects/test', methods=['GET'])
def test():
    rs = TaskInstance.query_task_instance()
    es_url = conf.get('elastic', 'url')
    es_user = conf.get('elastic', 'user')
    es_password = conf.get('elastic', 'password')
    index = 'idata_task_metrics-'
    suffix = time.strftime("%Y-%m", time.localtime())
    index = index + suffix
    es = Elasticsearch([es_url], http_auth=es_user+":"+es_password)
    datas = []
    for r in rs:
        r_j = {
            "type": r.type,
            "count": r.count,
            "cpu": r.cpu,
            "memory": r.memory,
            "date": int(time.time()*1000)
        }

        data = {
            "_index": index,
            "_type": "_doc",
            "_source": r_j
        }
        datas.append(data)
        if len(datas)>=3:
            helpers.bulk(es, datas)
            datas.clear()
    if len(datas)>0:
        helpers.bulk(es, datas)
        datas.clear()
    return jsonify(code=0, message='ok', data=None)
